
//Copyright 1997-2009 Syrinx Development, Inc.
//This file is part of the Syrinx Web Application Framework (SWAF).
// == BEGIN LICENSE ==
//
// Licensed under the terms of any of the following licenses at your
// choice:
//
//  - GNU General Public License Version 3 or later (the "GPL")
//    http://www.gnu.org/licenses/gpl.html
//
//  - GNU Lesser General Public License Version 3 or later (the "LGPL")
//    http://www.gnu.org/licenses/lgpl.html
//
//  - Mozilla Public License Version 1.1 or later (the "MPL")
//    http://www.mozilla.org/MPL/MPL-1.1.html
//
// == END LICENSE ==
using System;
using System.Collections;
using System.Threading;

namespace Swaf.Call
{
	/// <summary>
	/// Summary description for ThreadedCallbackQueue.
	/// </summary>
	public class ThreadedCallbackQueue : BaseCallbackQueue
	{
		private static IDictionary s_callbackThreadGroups = new Hashtable();
		private CallbackThreadGroup m_group;

		public static void shutdownThreadGroups()
		{
			foreach(CallbackThreadGroup g in s_callbackThreadGroups.Values)
				g.shutdown();
			s_callbackThreadGroups = new Hashtable();
		}
		public static void shutdownThreadGroups(IApplication app)
		{
			shutdownThreadGroups();
		}
		public ThreadedCallbackQueue(string groupName)
		{
			m_group = getGroup(groupName);
			m_group.addQueue(this);
		}

		public override void enqueueCall(ICall call)
		{
			base.enqueueCall (call);
			m_group.queuedCallAdded();
		}

		public override void enqueueCall(ICall call, params object[] info)
		{
			base.enqueueCall (call, info);
			m_group.queuedCallAdded();
		}



		private CallbackThreadGroup getGroup(string name)
		{
			lock(s_callbackThreadGroups)
			{
				if(s_callbackThreadGroups.Contains(name))
					return (CallbackThreadGroup)s_callbackThreadGroups[name];
				else
				{
					CallbackThreadGroup g = new CallbackThreadGroup(name);
					s_callbackThreadGroups[name] = g;
					return g;
				}
			}
		}

		private class CallbackThreadGroup
		{
			Thread m_thread;
			IList m_queues;
			Semaphore m_wait;
			string m_name;
			object syncObj = new object();
			bool shutDown = false;

			public CallbackThreadGroup(string name)
			{
				m_name = name;
				m_wait = new Semaphore(0);
				m_queues = new ArrayList();
				m_thread = new Thread(new ThreadStart(processQueues));
				m_thread.Start();
			}

			public bool IsShutdown
			{
				get
				{
					lock (syncObj)
					{
						return shutDown;
					}
				}
				set
				{
					lock (syncObj)
					{
						shutDown = value;
					}
				}
			}

			public void queuedCallAdded()
			{
				m_wait.release();
			}

			public void addQueue(ThreadedCallbackQueue q)
			{
				m_queues.Add(q);
			}

			private void processQueues()
			{
				while (! IsShutdown)
				{
					m_wait.acquireAll(); //Waits for a queue to say it has work.
                    if (IsShutdown) break;

					int count = m_queues.Count;
					for (int pos = 0; pos < count; ++pos)
					{
						try
						{
							ThreadedCallbackQueue q = m_queues[pos] as ThreadedCallbackQueue;
							q.processQueue();
						}
						catch (Exception e)
						{
							//Just log the exception and keep processing the queue.  We cannot let a failure in
							//processing a queue kill this thread.
							Application.globalApp.Log.write("Exception working with threadedCallbackqueue in ThreadedCallbackQueue.processQueues: {0}", e.ToString());
						}
					}
				}
			}

			public void shutdown()
			{
				IsShutdown = true;

                // mark queue to shutdown
                foreach (ThreadedCallbackQueue q in m_queues)
                {
                    q.IsShutdown = true;
                }

				// Ensure thread wakes up to end
				m_wait.release();
			}
		}
	}
}
